package com.darrenchan.spark

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * @Desc
  * @Author chenchi03
  * @CreateTime 2019-10-27 16:14
  */
object CheckPoint {
    def main(args: Array[String]): Unit = {

        val checkpointDirectory = "file:///Users/chenchi03/data"

        def functionToCreateContext(): StreamingContext = {
            val sparkConf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount2")
            val ssc = new StreamingContext(sparkConf, Seconds(20))   // new context
            val lines = ssc.socketTextStream("localhost", 9999) // create DStreams


            val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)

            result.print()


            ssc.checkpoint(checkpointDirectory)   // set checkpoint directory
            ssc
        }


        val ssc = StreamingContext.getOrCreate(checkpointDirectory, () => functionToCreateContext())
        ssc.start()
        ssc.awaitTermination()

    }
}
